package studio.raptor.extractor.segment.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import studio.raptor.databus.client.DatabusConfig;
import studio.raptor.databus.client.DatabusProducer;
import studio.raptor.databus.client.DatabusProducerBuilder;
import studio.raptor.databus.command.*;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.BuildJsonResult;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.service.EventMetaService;

import java.sql.SQLException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by liuyong8 on 2017/7/28.
 */
@Component
public class EventDeliveringTask implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(EventDeliveringTask.class);
    private BlockingQueue<BuildJsonResult> buildJsonQueue;
    private ServerConfig serverConfig;
    private int count;
    private DatabusProducer producer;
    @Autowired
    private EventMetaService eventMetaService;
    @Override
    public void run() {

        try {
            producer = createProducer();
        } catch (Exception e) {
            e.printStackTrace();
        }

        while (!Thread.currentThread().isInterrupted()){
            BuildJsonResult buildJsonResult = null;
            try {
                buildJsonResult = buildJsonQueue.poll(10, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            if (buildJsonResult == null){
                printPerfTip();
                continue;
            }
            EventMeta eventMeta = buildJsonResult.getEventMeta();
            try {//发送报文
                EventBasicData eventBasicData = cover2EventBasicData(eventMeta);
                EventBusinessData eventBusinessData = cover2EventBusinessData(buildJsonResult);
                EventMetaData eventMetaData = cover2EventMetaData(eventMeta);
                Event event = new Event(eventBasicData,eventBusinessData,eventMetaData);
                producer.send(event);
                log.debug("send event {}",event.toString());

                //更新event 为 C
                eventMetaService.modifyEventMetaStatus2C(eventMeta);

            } catch (Exception e) {
                try {
                    eventMetaService.modifyEventMetaStatus2F(eventMeta);
                    e.printStackTrace();
                } catch (SQLException e1) {
                    e1.printStackTrace();
                }
            }



        }

    }

    private DatabusProducer createProducer() throws Exception{
        DatabusConfig config = new DatabusConfig(serverConfig.getBizName());
        DatabusProducerBuilder producerBuilder = new DatabusProducerBuilder(config);
        DatabusProducer producer = producerBuilder.create();
        return producer;
    }

    private EventBasicData cover2EventBasicData(EventMeta eventMeta){
        EventBasicData eventBasicData = new EventBasicData();
        eventBasicData.setName(serverConfig.getEventName() + "_data");
        eventBasicData.setSource(serverConfig.getBizName());
        eventBasicData.setKindId(8002);
        eventBasicData.setDeliverTime(eventMeta.getDeliverTime());
        return  eventBasicData;
    }

    private EventBusinessData cover2EventBusinessData(BuildJsonResult buildJsonResult){
        EventBusinessData eventBusinessData = new EventBusinessData();
        eventBusinessData.setContent(buildJsonResult.getJson());
        eventBusinessData.setDataService("1");
        eventBusinessData.setDataType(EventBusinessDataType.DATATYPE_JSON);
        eventBusinessData.setProvideType(EventBusinessDataProvideType.PROVIDE_TYPE_CARRY);
        return eventBusinessData;
    }

    private EventMetaData cover2EventMetaData(EventMeta eventMeta){
        EventMetaData eventMetaData = new EventMetaData();
//        private String businessNo;
        eventMetaData.setBusinessNo(eventMeta.getObjectKey());
//        private EventMetaDataBusinessType businessType;
        eventMetaData.setBusinessType(eventMeta.getEventType() == 0 ? EventMetaDataBusinessType.BUSINESS_TYPE_NRM : EventMetaDataBusinessType.BUSINESS_TYPE_SAR);
//        private Long regionId;
        eventMetaData.setRegionId(eventMeta.getRegionId());
//        private String prodIds;
        eventMetaData.setProdIds(eventMeta.getProdIds());
//        private String offerIds;
        eventMetaData.setOfferIds(eventMeta.getOfferIds());
//        private String offerSpecAttrs;
        eventMetaData.setOfferSpecAttrs(eventMeta.getOfferSpecAttr());
//        private String serviceOfferIds;
        eventMetaData.setServiceOfferIds(eventMeta.getServiceOfferIds());
//        private String serviceSpecAttrs;
        eventMetaData.setServiceSpecAttrs(eventMeta.getServiceSpecAttr());
//        private Long[] offerProdId;
        eventMetaData.setOfferProdId(null);
//        private String orderTypeIds;
        eventMetaData.setOrderTypeIds(eventMeta.getOrderTypeId());
//        private Long[] custId;
        eventMetaData.setCustId(eventMeta.getCustId());
//        private Long[] acctId;
        eventMetaData.setAcctId(eventMeta.getAcctId());
//        private Long[] prodInstId;
        eventMetaData.setProdInstId(eventMeta.getProdInstId());
//        private Long[] offerInstId;
        eventMetaData.setOfferInstId(eventMeta.getOfferInstId());
        return eventMetaData;
    }

    private void printPerfTip(){
        if ((++count) % 1000 == 0) {
            log.warn("builded events have been delivered, please speed up the building");
            count = 0;
        }
    }

    public void setBuildJsonQueue(BlockingQueue<BuildJsonResult> buildJsonQueue) {
        this.buildJsonQueue = buildJsonQueue;
    }

    public void setServerConfig(ServerConfig serverConfig) {
        this.serverConfig = serverConfig;
    }
}
